• 繼續看一下 channelFlow

運用場景

• 可以從多個來源同步地處裡 data stream.
• 可以使用 flow 實作 producer/consumer 模式.
• 可以使用非同步操作建立複雜的 data pipeline建立資料管道.

examples

結合多個 channel

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val channel1 = Channel<Int>()
    val channel2 = Channel<Int>()

    launch {
        for (i in 1..50) {
            delay(100)
            channel1.send(i)
        }
        channel1.close()
    }

    launch {
        for (i in 6..10) {
            delay(120)
            channel2.send(i)
        }
        channel2.close()
    }

    // use channelFlow combines two channel's data stream 
    val combinedFlow = channelFlow {
        launch { 
            for (value in channel1) {
                send(value)
            }
        }
        
        launch { 
            for (value in channel2) {
                send(value)
            }
        }
    }

    combinedFlow.collect { value ->
        println("Collected data: $value")
    }
}

處裡事件的 stream

import kotlinx.coroutines.*
import kotlinx.coroutines.channels.*
import kotlinx.coroutines.flow.*

fun main() = runBlocking {
    val eventsChannel = Channel<String>()

    launch {
        repeat(10) {
            delay(300)
            eventsChannel.send("Event $it")
        }
        eventsChannel.close() // must Close the channel when done
    }

    val eventsFlow = channelFlow<String> {
        for (event in eventsChannel) {
            send("Processed: $event")
        }
    }

    eventsFlow.collect { processedEvent ->
        println("collected: " + processedEvent)
    }
}

注意事項

• 使用 channelFlow 時, 在 suspend lambda中有是當的取消處理，這是要避免 leak 或是意外的行為.

• 恩, 有點趕